package com.bossien.flink.handler.exam.handler;

import com.alibaba.fastjson.JSONObject;
import com.bossien.common.model.student.ExamQuestionAnswer;
import com.bossien.flink.config.RMQExcutionStreamHandlerConfig;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.stereotype.Component;

/**
 * @author 胡春林
 * 考试流执行器
 */
@Slf4j
@Accessors(chain = true)
@Component
public class ExamDataStreamExecute extends RMQExcutionStreamHandlerConfig<DataStream> {

    @Override
    public DataStream<Tuple2<String,ExamQuestionAnswer>> executeFilter(DataStream dataStream) {

        //转换参数
     log.info(dataStream.print().getTransformation().getInput().getName());

        SingleOutputStreamOperator<Tuple2<String,ExamQuestionAnswer>> keyedStream = dataStream.map(new ExamMap())
        .keyBy(0).timeWindow(Time.seconds(30))
        .apply(new WindowFunction<Tuple2<String, ExamQuestionAnswer>, Tuple2<String, String>, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Tuple2<String, ExamQuestionAnswer>> iterable, Collector<Tuple2<String, String>> collector) throws Exception {

                int  sum = 0;
                for(Tuple2<String, ExamQuestionAnswer> record : iterable) {
                    sum += record.f1.getScore();
                }

                //获取单个元素
                Tuple2<String,ExamQuestionAnswer> eache = iterable.iterator().next();
                Tuple2<String,String> result = new Tuple2<>();
                result.f0 =eache.f0;
                result.f1 = String.valueOf(sum);

                log.info("该学员成绩："+result.f1);
                collector.collect(result);
            }
        });

        return  keyedStream;
    }

    @Override
    public DataStream addSkin(DataStream tClass) {
        return null;
    }

    public class ExamMap implements  MapFunction<String,Tuple2<String,ExamQuestionAnswer>>
    {
        @Override
        public Tuple2<String, ExamQuestionAnswer> map(String s) throws Exception {

            ExamQuestionAnswer examQuestionAnswer = JSONObject.parseObject(s, ExamQuestionAnswer.class);

            Tuple2<String,ExamQuestionAnswer> tuple2 = new Tuple2<>();
            tuple2.f0 = String.valueOf(examQuestionAnswer.getUserId());
            tuple2.f1 = examQuestionAnswer;
            return tuple2;
        }
    }
}
